﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Messaging;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;


/// <summary>
/// 使用Task实现
/// </summary>
public class MessageQueueTaskPool<T> where T:class {
  public static readonly log4net.ILog logger = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
  /// <summary>
  /// 连接失败后重试秒数
  /// </summary>
  public int RetryTimeout = 10;
  /// <summary>
  /// 接收消息等待秒数，每过这些秒数会重启一次
  /// 同样的当Stop时，也会有大概这些秒数的延迟
  /// </summary>
  public int ReceiveTimeout = 10;
  private IMessageFormatter formatter;
  public Action<string> HandleLog = (s) => {
    logger.Info(s);
  };
  public string QueueName {
    set {
      this.queueName = value;
      this.QueuePath = @".\Private$\" + this.queueName;
    }
    get => this.queueName;
  }
  private string queueName;
  private string QueuePath { get; set; }

  public MessageQueueTaskPool() { 
    this.formatter = new XmlMessageFormatter(new Type[] { typeof(T) });
  }
  public MessageQueueTaskPool(string queueName){
    this.QueueName = queueName;
    this.formatter = new XmlMessageFormatter(new Type[] { typeof(T) });
  }
  public MessageQueueTaskPool(IMessageFormatter formatter, string queueName) {
    this.QueueName = queueName;
    this.formatter = formatter;
  }
  public MessageQueueTaskPool(IMessageFormatter formatter) {
    this.formatter = formatter;
  }

  /// <summary>
  /// 收到消息事件
  /// </summary>
  public event HandleMessageDelegate OnMessage;
  public class MessageEventArgs {
    public T data;
    public long count;
    public string name;
    public string id;
  }
  public delegate void HandleMessageDelegate(MessageEventArgs evt);

  /// <summary>
  /// receiver列表
  /// </summary>
  private Dictionary<string, Worker> receiverList = new Dictionary<string, Worker>();
  private const string MESSAGE_SENDER = "__MESSAGE_SENDER__";
  private Worker _sender;
  private Worker Sender {
    get {
      if (_sender == null) {
        _sender = new Worker(this, MESSAGE_SENDER, true);
      }
      return _sender;
    }
  }
  
  /// <summary>
  /// 推送消息 
  /// </summary>
  public bool SendMessage(T msg) {
    return Sender.pushMessage(msg);
  }
  /// <summary>
  /// 推送消息 
  /// </summary>
  public bool SendMessage(string name,T msg) {
    if (receiverList.ContainsKey(name)) {
      return receiverList[name].pushMessage(msg);
    }
    return false;
  }

  /// <summary>
  /// 开启Worker
  /// 当Worker不存在时，会自动新建
  /// </summary>
  public void StartReceive(string name) {
    Worker w = AddReceiver(name);
    w.startReceive();
  }

  /// <summary>
  /// 结束Workder
  /// </summary>
  public bool StopReceive(string name) {
    if (receiverList.ContainsKey(name)) {
      Worker w = receiverList[name];
      w.stopReceive();
      return true;
    }
    return false;
  }
  public bool RemoveWorker(string name) {
    if (receiverList.ContainsKey(name)) {
      var w = receiverList[name];
      w.stopReceive();
      w.Dispose();
    }
    return false;
  }

  /// <summary>
  /// 判断Worker是否正在运行
  /// </summary>
  public bool IsRuning(string name) {
    if (receiverList.ContainsKey(name)) {
      Worker w = receiverList[name];
      return w.IsRuning;
    }
    return false;
  }

  /// <summary>
  /// 添加worker
  /// </summary>
  public Worker AddReceiver(string name) {
    Worker w;
    if (receiverList.ContainsKey(name)) {
      w = receiverList[name];
    } else {
      w = new Worker(this, name, false);
      receiverList.Add(name, w);
    }
    return w;
  }



  public class Worker : IDisposable {
    public string Name { get; set; }
    public Worker(MessageQueueTaskPool<T> ctx, string name, bool isSender) {
      this.ctx = ctx;
      Name = name;
    }
    ~Worker() {
      this.Dispose();
    }
    
    private MessageQueueTaskPool<T> ctx;
    private MessageQueue queue;
    private Task receiveTask;
    private CancellationTokenSource tokenSource;
    public long count;

    /// <summary>
    /// 处理消息
    /// </summary>
    private void handleMessage(long count, Message msg) {
      ctx.OnMessage?.Invoke(new MessageEventArgs() {
        data = msg.Body as T,
        count = count,
        name = Name,
        id = msg.Id
      });
    }
    /// <summary>
    /// 推送消息
    /// </summary>
    public bool pushMessage(T obj) {
      if (initQueue()) {
        Message msg = new Message(obj, ctx.formatter);
        try {
          queue.Send(msg);
          return true;
        } catch (Exception err){
          ctx.HandleLog("==> MQ " + Name + " 发送失败:" + err.Message);
        }
      }
      return false;
    }


    public bool IsRuning {
      get {
        return receiveTask != null && !receiveTask.IsCanceled && !receiveTask.IsCompleted;
      }
    }
    public void startReceive() {
      ctx.HandleLog("==> MQ " + Name + " 开始接收");
      if (!IsRuning) {
        //初始化
        tokenSource = new CancellationTokenSource();
        count = 0;

        receiveTask = new Task(async () => {
          while (true) {
            if (tokenSource.Token.IsCancellationRequested) {
              break;
            }
            receiveMessage(++count);
          }
          receiveTask = null;
        }, tokenSource.Token);
        receiveTask.Start();
      } else {
        ctx.HandleLog("==> MQ " + Name + " 开始接收(已经开启)");
      }
    }
    public void stopReceive() {
      if (IsRuning) {
        ctx.HandleLog("==> MQ " + Name + " 停止接收");
        this.tokenSource.Cancel();
        this.queue.Close();
      } else {
        ctx.HandleLog("==> MQ " + Name + " 停止接收(未启动)");
      }
    }

    private bool initQueue() {
      if (tokenSource != null && tokenSource.Token.IsCancellationRequested) {
        return false;
      }
      //消息队列路径
      string path = this.ctx.QueuePath;
      if (string.IsNullOrEmpty(path)) {
        ctx.HandleLog("==> MQ " + Name + " 初始化失败：必须提供队列路径");
        return false;
      }
      try {
        if (this.queue == null || this.queue.Path != path) {
          // 如果存在指定路径的消息队列
          if (MessageQueue.Exists(path)) {
            //获取这个消息队列
            queue = new MessageQueue(path);
          } else {
            //不存在，就创建一个新的，并获取这个消息队列对象
            queue = MessageQueue.Create(path);
          }
          queue.Formatter = ctx.formatter;
          ctx.HandleLog("==> MQ " + Name + " 连接成功");
        }
      } catch {
        return false;
      }
      return true;
    }

    /// <summary>
    /// 接收消息
    /// </summary>
    private void receiveMessage(long count) {
      if (initQueue()) {
        Message msg;
        try {
          msg = queue.Receive(new TimeSpan(ctx.ReceiveTimeout * 1000L * 10000));
          handleMessage(count, msg);
        } catch (MessageQueueException err) {
          if (err.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout) {
            ctx.HandleLog("==> MQ " + Name + " 出现错误(" + ctx.RetryTimeout + "秒后重试): " + err.Message);
            Thread.Sleep(ctx.RetryTimeout * 1000);
          }
        }
      } else { //不成功，n秒重试
        ctx.HandleLog("==> MQ " + Name + " 连接MessageQueue失败(" + ctx.RetryTimeout + "秒后重试)");
        Thread.Sleep(ctx.RetryTimeout * 1000);
      }
    }

    public void Dispose() {
      try {
        this.stopReceive();
        if (tokenSource != null) {
          tokenSource.Dispose();
        }
        if (receiveTask != null) {
          receiveTask.Dispose();
        }
        if (queue != null) {
          queue.Dispose();
        }
      } catch { }
    }
  }











}
